package Autodesk;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Hashtable;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class SessionFilter extends Configured implements Tool  {
	
		public Path prodPath;

		public static class LogMapper extends Mapper<Object, Text, Text, Text> {

			// private final static IntWritable one = new IntWritable(1);
			private Text word = new Text();
			Hashtable<String, String> ht = new Hashtable<String, String>();
		@Override
		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {

			String record[] = value.toString().toLowerCase().split("\t", 12);
			StringBuffer bf = new StringBuffer();
			StringBuffer bv=new StringBuffer();
			try{
			bf.append(record[0]);// DATE
			bf.append("\t");
			bf.append(record[2]);//ipPort
			word.set(bf.toString());
			
			bv.append(record[5]);//Hour
			bv.append("\t");
			bv.append(record[4]);// Language
			bv.append("\t");
			bv.append(record[3]);// Product
			bv.append("\t");
			bv.append(record[1]);// Channel
			bv.append("\t");
			bv.append(record[8]);// Country
			Text out=new Text();
			out.set(bv.toString());
			
			//context.write(word, output);
			context.write(word, out);
//			output.collect(word,out);
			}catch(Exception e){
				System.out.println("Error Reg: "+value);
				System.out.println("PROBLEM: "+e.getMessage());
			}
			
		}
	}
		public static class LogReducer extends
		Reducer<Text, Text, Text, IntWritable> {
//	public static class LogReducer extends MapReduceBase implements Reducer<Text, Text, Text, IntWritable> {
		private IntWritable result = new IntWritable(1);
		public void reduce(Text key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException {
//		@Override
//		public void reduce(Text key, Iterator<Text> values,
//				OutputCollector<Text, IntWritable> output, Reporter arg3)
//				throws IOException {
			ReduceHours red=new ReduceHours();
			ArrayList<String> listValues=new ArrayList<String>();
			ArrayList<String> listHours=new ArrayList<String>();
			ArrayList<ArrayList<String>> listSessions=new ArrayList<ArrayList<String>>();
			ArrayList<String> sessionIdentyfied=new ArrayList<String>();
			listValues=red.makeList(values.iterator());
			listHours=red.generateListHours(listValues);	
			if(!listHours.isEmpty()){
				red.orderList(listHours);
				listSessions=red.buildListOfSessions(listHours);
				for(int g=0;g<listValues.size();g++){
					String value=listValues.get(g).toString();
					String[] line=value.split("\t");
					sessionIdentyfied=red.identifySession(listSessions, line[0]);
					String newKey=key.toString()+"\t"+line[1]+"\t"+line[2]+"\t"+line[3]+"\t"
					+sessionIdentyfied.get(0)+"\t"+sessionIdentyfied.get(1)+"\t"+line[4];
					//output.collect(new Text(newKey), result);
					context.write(new Text(newKey), result);
				}
				
			}
		}
	}

	
	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "ParseLogs");
		job.setJarByClass(Autodesk.SessionFilter.class);
		
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job,new Path(args[1]));
		
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		System.exit(job.waitForCompletion(true)?0:1);

		return 0;
	
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new SessionFilter(), args);
		System.exit(res);
		}

 
}